package com.bright.ghj.overt.service.impl;

import com.bright.ghj.common.config.BizThreadPoolExt;
import com.bright.ghj.common.pojo.dto.PageDTO;
import com.bright.ghj.common.pojo.vo.PageVO;
import com.bright.ghj.common.util.CollectionUtil;
import com.bright.ghj.common.util.PageUtil;
import com.bright.ghj.overt.converter.AsyncMessageConverter;
import com.bright.ghj.overt.enums.AsyncTaskExecInfo;
import com.bright.ghj.overt.manager.AsyncMessageManager;
import com.bright.ghj.overt.pojo.dto.AsyncMessageDTO;
import com.bright.ghj.overt.pojo.query.AsyncTaskMassageQuery;
import com.bright.ghj.overt.pojo.vo.AsyncMessageVO;
import com.bright.ghj.overt.pojo.vo.AsyncTaskExecInfoResultVO;
import com.bright.ghj.overt.pojo.vo.AsyncTaskResultVO;
import com.bright.ghj.overt.service.AsyncMessageService;
import com.bright.ghj.overt.task.timely.AsyncTaskExecutor;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import static com.bright.ghj.overt.enums.AsyncTaskExecInfo.EXECUTING;
import static com.bright.ghj.overt.enums.AsyncTaskExecInfo.TODO_EXEC;

/**
 * @author Tz
 * @date 2024/01/22 15:03
 * @Description 异步任务消息接口 实现
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class AsyncMessageServiceImpl implements AsyncMessageService {

    private final AsyncMessageManager asyncMessageManager;
    private final AsyncMessageConverter asyncMessageConverter;

    /**
     * 保存执行任务对象，可以获取进度等信息
     */
    public static ConcurrentMap<String, AsyncTaskExecutor> CONCURRENT_MAP = new ConcurrentHashMap<>();

    /**
     * 异步任务执行的扩展线程池
     */
    private final BizThreadPoolExt bizPool;

    /**
     * 定时任务执行的扩展线程池
     */
    private final BizThreadPoolExt jobPool;

    @Override
    public AsyncMessageDTO save(AsyncMessageDTO asyncMessageDTO) {
        return asyncMessageManager.save(asyncMessageDTO);
    }

    /**
     * 根据id查询异步任务消息
     *
     * @param id 主键
     * @return   异步任务消息DTO
     */
    @Override
    public AsyncMessageDTO getById(Integer id) {
        return asyncMessageManager.getById(id);
    }


    @Override
    public AsyncTaskResultVO execAsyncTask(AsyncTaskExecutor asyncTaskExecutor) {

        AsyncTaskResultVO asyncTaskResultVO = new AsyncTaskResultVO();

        boolean checkAsyncTaskExecutor = asyncTaskExecutor.checkAsyncTaskExecutor();
        if (!checkAsyncTaskExecutor) {
            //提交任务失败
            asyncTaskResultVO.setIsSubmitTask(false);
            asyncTaskResultVO.setNoSubmitIllustrate("执行器不通过校验,可能的原因: 未设置同步锁或未识别的执行器");
            return asyncTaskResultVO;
        }

        asyncTaskResultVO.setKeyword(asyncTaskExecutor.getKeyword());

        //保存执行记录
        AsyncMessageDTO save = save(asyncTaskExecutor);
        if (save != null) {
            asyncTaskExecutor.setId(save.getId());
            CompletableFuture.runAsync(() -> {
                        Thread.currentThread().setName(asyncTaskExecutor.getTaskName() + "_" + asyncTaskExecutor.getKeyword());
                        asyncTaskExecutor.baseExec();
                    }, bizPool)
                    .exceptionally(throwable -> {
                        throwable.printStackTrace();
                        return null;
                    });

            //保存执行的任务
            CONCURRENT_MAP.put(asyncTaskExecutor.getKeyword(), asyncTaskExecutor);

            //提交任务成功
            asyncTaskResultVO.setIsSubmitTask(true);

        } else {

            //提交任务失败
            asyncTaskResultVO.setIsSubmitTask(false);
        }

        return asyncTaskResultVO;
    }

    /**
     * 根据主键标识获取异步任务的进度
     * <p>
     * 当项目重启 如果获取历史执行任务的进度会报错 直接返回100
     * <p>
     * 添加一个可以自动清理已经完成任务的功能 避免一直占用内存 {@link AsyncMessageServiceImpl#cleanAsyncTaskTimer()}
     * @param keyword 主键标识
     * @return        进度百分比
     */
    @Override
    public Integer getCompletionDegree(String keyword) {

        //如果不存在 可能是任务完成后已经被清除了
        if (!CONCURRENT_MAP.containsKey(keyword)) {
            //直接返回100
            return 100;
        }
        return CONCURRENT_MAP.get(keyword).getCompletionDegree();
    }

    /**
     * 根据id撤回发送的异步任务
     *
     * @param id 主键
     * @return 返回撤回的数量
     */
    @Override
    public Integer revokeAsyncTask(Integer id) {
        Integer resultSize = asyncMessageManager.revokeAsyncTask(id);
        if (resultSize <= 0) {
            throw new RuntimeException("没有可撤回的任务或当前任务正在执行中！");
        }
        return resultSize;
    }

    /**
     * 根据条件获取异步任务消息列表
     *
     * @param asyncTaskMassageQuery 查询条件
     * @return                      异步消息列表
     */
    @Override
    public PageVO<AsyncMessageVO> pageAsyncMessage(AsyncTaskMassageQuery asyncTaskMassageQuery) {

        PageDTO<AsyncMessageDTO> pageDTO = asyncMessageManager.pageAsyncMessage(asyncTaskMassageQuery, PageUtil.toPageable(asyncTaskMassageQuery));
        PageVO<AsyncMessageVO> pageVO = new PageVO<>();
        pageVO.setData(asyncMessageConverter.dtoConvertVO(pageDTO.getData()));
        pageVO.setTotal(pageDTO.getTotal());
        pageVO.setPageCount(pageDTO.getPageCount());
        return pageVO;
    }

    /**
     * 根据keyword集合查询异步消息列表
     *
     * @param keywords keyword集合
     * @return         异步消息列表
     */
    @Override
    public List<AsyncMessageVO> listAsyncMessageByKeywords(List<String> keywords) {
        List<AsyncMessageDTO> asyncMessageDTOS = asyncMessageManager.listAsyncMessageByKeywords(keywords);
        return asyncMessageConverter.dtoConvertVO(asyncMessageDTOS);
    }

    /**
     * 一键已读
     *
     * @param statusCategory 需要已读的状态类别
     * @param username       需要已读的用户
     * @return 已读记录数
     */
    @Override
    public Integer oneClickRead(Integer statusCategory, String username) {
        //状态类别需要转实际的状态
        List<Integer> execStatusList = AsyncTaskExecInfo.getStatusFromCategory(statusCategory);
        return asyncMessageManager.oneClickRead(execStatusList, username);
    }

    /**
     * 阅读一条未读消息
     *
     * @param id 主键
     * @return 读取结果
     */
    @Override
    public Integer read(Integer id) {
        return asyncMessageManager.read(id);
    }

    /**
     * 获取异步任务执行的情况结果
     * @param username 执行任务的用户名
     * @return         异步任务的执行情况结果的VO
     */
    @Override
    public AsyncTaskExecInfoResultVO getAsyncTaskExecInfo(String username) {

        //TODO 集群情况下需要考虑换种方式

        AsyncTaskExecInfoResultVO asyncTaskExecInfoResultVO = new AsyncTaskExecInfoResultVO();
        asyncTaskExecInfoResultVO.setIsExistsCompleteTask(false);
        asyncTaskExecInfoResultVO.setCompleteTaskMessages(new ArrayList<>());

        Map<String, Integer> taskCount = new HashMap<>();

        //循环遍历所有提交的任务
        Collection<AsyncTaskExecutor> asyncTaskExecutors = CONCURRENT_MAP.values();
        for (AsyncTaskExecutor asyncTaskExecutor : asyncTaskExecutors) {
            if (asyncTaskExecutor.getUsername().equalsIgnoreCase(username)
                    && !asyncTaskExecutor.getIsRunning()) {
                asyncTaskExecInfoResultVO.setIsExistsCompleteTask(true);
                taskCount.put(asyncTaskExecutor.getTaskType(), taskCount.getOrDefault(asyncTaskExecutor.getTaskType(), 0) + 1);
            }
        }

        //纪录执行任务的情况
        if (!CollectionUtil.isEmpty(taskCount) && asyncTaskExecInfoResultVO.getIsExistsCompleteTask()) {
            for (String key : taskCount.keySet()) {
                asyncTaskExecInfoResultVO.getCompleteTaskMessages().add(key + ",有完成的任务!");
            }
        }
        return asyncTaskExecInfoResultVO;
    }

    /**
     * 确认读取执行的结果
     *
     * @param username 执行任务的用户名
     */
    @Override
    public void confirmReadingExecResult(String username) {
        //TODO 集群情况下需要考虑换种方式
        //循环遍历所有提交的任务
        Collection<AsyncTaskExecutor> asyncTaskExecutors = CONCURRENT_MAP.values();
        for (AsyncTaskExecutor asyncTaskExecutor : asyncTaskExecutors) {
            if (asyncTaskExecutor.getUsername().equalsIgnoreCase(username)
                    && !asyncTaskExecutor.getIsRunning()) {
                CONCURRENT_MAP.remove(asyncTaskExecutor.getKeyword());
            }
        }
    }


    /**
     * 清理异步消息的定时器
     * <p>
     * initialDelay = 300000: 延迟执行时间 单位毫秒 这里是5分钟
     * <p>
     * fixedRate = 1800000:  执行间隔时间 单位毫秒 这里是30分钟
     * <p>
     * Scheduled(initialDelay = 3000, fixedRate = 3000): 测试时用的
     */
    @Scheduled(initialDelay = 300000, fixedRate = 1800000)
    public void cleanAsyncTaskTimer() {
        List<AsyncMessageVO> asyncMessageVOS = listAsyncMessageByKeywords(new ArrayList<>(CONCURRENT_MAP.keySet()));
        if (!CollectionUtil.isEmpty(asyncMessageVOS)) {
            for (AsyncMessageVO asyncMessageVO : asyncMessageVOS) {
                //清除不是待处理和正在运行的任务
                if (!asyncMessageVO.getExecStatus().equals(TODO_EXEC.getStatus())
                        && !asyncMessageVO.getExecStatus().equals(EXECUTING.getStatus())) {
                    log.info("清除keyword:[{}], 异步任务的缓存", asyncMessageVO.getKeyword());
                    CompletableFuture.runAsync(() -> {
                        Thread.currentThread().setName("clean_task_" + asyncMessageVO.getTaskName() + "_" + asyncMessageVO.getKeyword());
                        CONCURRENT_MAP.remove(asyncMessageVO.getKeyword());
                    }, jobPool);
                }
            }
        }
    }
}
